package org.example.base.kafka;

import lombok.extern.slf4j.Slf4j;
import org.example.exceptions.BusinessException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;

import javax.annotation.Resource;
import java.util.concurrent.CompletableFuture;

import static org.example.enums.ErrorCodeEnum.KAFKA_MSG_SEND_EXCEPTION;

@Service
@Slf4j
public class KafkaMsgSendProducerBase<K, V> {
    @Resource(name = "customKafkaTemplate")
    @Lazy
    private KafkaTemplate<K, V> kafkaTemplate;

    public CompletableFuture<SendResult<K, V>> sendMsg(String topic, V msg) {
        return kafkaTemplate.send(topic, msg).completable();
    }

    public CompletableFuture<SendResult<K, V>> sendMsg(String topic, K key, V msg) {
        return kafkaTemplate.send(topic, key, msg).completable();
    }

    public CompletableFuture<SendResult<K, V>> sendMsg(String topic, Integer partition, K key, V msg) {
        return kafkaTemplate.send(topic, partition, key, msg).completable();
    }

    public CompletableFuture<SendResult<K, V>> sendMsg(String topic, Integer partition, K key, V msg, Long timestamp) {
        return kafkaTemplate.send(topic, partition, timestamp, key, msg).completable();
    }

    public SendResult<K, V> sendMsgSync(String topic, V msg) {
        try {
            return kafkaTemplate.send(topic, msg).get();
        } catch (Exception e) {
            log.error("kafka 消息发送失败 msg={}, topic={}", msg.toString(), topic, e);
            throw new BusinessException(KAFKA_MSG_SEND_EXCEPTION);
        }
    }

    /**
     * 异步发送消息
     * @param topic
     * @param msg
     * @param successCallback 成功回调
     * @param failureCallback 失败回调
     */
    public void sendMsgAsync(String topic, V msg, SuccessCallback<SendResult<K, V>> successCallback, FailureCallback failureCallback) {
        kafkaTemplate.send(topic, msg).addCallback(
                successCallback,
                failureCallback
        );
    }
}
